package com.talent.service.iot.mqtt.service;

import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @ClassName：DefaultMqttRetainMsgServiceImpl
 * @Author: hszhou
 * @Date: 2025/1/8 14:17
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalMqttRetainMsgServiceImpl implements IMqttRetainMsgService,ITopicService{

    private static final Logger logger = LoggerFactory.getLogger(LocalMqttRetainMsgServiceImpl.class);


    public static final ConcurrentHashMap<String, TlMqttPublishReq> RETAIN_MAP = new ConcurrentHashMap<>();
    @Override
    public Mono<TlMqttPublishReq> save(String topic, TlMqttPublishReq req) {

        return Mono.justOrEmpty(RETAIN_MAP.put(topic, req));
    }

    @Override
    public Flux<TlMqttPublishReq> find(String topic) {

        List<TlMqttPublishReq> list = RETAIN_MAP.entrySet().stream()
                .filter(entry -> match(entry.getKey(), topic))
                .map(Map.Entry::getValue)
                .toList();
        return Flux.fromIterable(list);

    }
    @Override
    public void clear(String topic) {
        RETAIN_MAP.remove(topic);

    }
}
